{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# encoding=utf-8\n",
    "\n",
    "from urllib.parse import urljoin\n",
    "\n",
    "from bs4 import BeautifulSoup\n",
    "from tornado import gen, httpclient, ioloop, queues\n",
    "\n",
    "base_url = 'http://www.tornadoweb.org/en/stable/'\n",
    "concurrency = 3\n",
    "\n",
    "\n",
    "async def get_url_links(url):\n",
    "    response = await httpclient.AsyncHTTPClient().fetch('http://www.tornadoweb.org/en/stable/')\n",
    "    html = response.body.decode('utf-8')\n",
    "    soup = BeautifulSoup(html)\n",
    "    links = [urljoin(base_url, a.get('href')) for a in soup.find_all('a', href=True)]\n",
    "    return links\n",
    "\n",
    "\n",
    "async def main():\n",
    "    ## 记录已经抓取的url\n",
    "    seen_set = set()\n",
    "    ## tornado下的queues是支持异步非阻塞的队列\n",
    "    q = queues.Queue()\n",
    "\n",
    "    async def fetch_url(current_url):\n",
    "        # 生产者\n",
    "        if current_url in seen_set:\n",
    "            return\n",
    "\n",
    "        print('获取：{}'.format(current_url))\n",
    "        seen_set.add(current_url)\n",
    "        next_urls = await get_url_links(current_url)\n",
    "        for new_url in next_urls:\n",
    "            # 判断是否是该域名下的链接\n",
    "            if new_url.startswith(base_url):\n",
    "                ## 放入队列，队列计数器+1\n",
    "                await q.put(new_url)\n",
    "\n",
    "    async def worker():\n",
    "        async for url in q:\n",
    "            # 消费者判断退出\n",
    "            if url is None:\n",
    "                return\n",
    "\n",
    "            try:\n",
    "                await fetch_url(url)\n",
    "            except Exception as e:\n",
    "                print('exception')\n",
    "            finally:\n",
    "                ## 处理后，队列计数器-1\n",
    "                q.task_done()\n",
    "\n",
    "    # 1、放入初始url到队列\n",
    "    await q.put(base_url)\n",
    "\n",
    "    # 2、启动协程，同时初始化3个协程\n",
    "    workers = gen.multi([worker() for _ in range(concurrency)])\n",
    "    await q.join()\n",
    "\n",
    "    for _ in range(concurrency):\n",
    "        await q.put(None)\n",
    "\n",
    "    await workers\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    io_loop = ioloop.IOLoop.current()\n",
    "    io_loop.run_sync(main)\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
